 1.任务调度系统Airflow之Airflow安装部署
   
   1).安装依赖
   CentOS 7.X
   Python 3.5 或以上版本（推荐）
   MySQL 5.7.x
   Apache-Airflow 1.10.11
   虚拟机可上网，需在线安装包
   备注：后面要安装的三个软件Airflow、Atlas、Griffin，相对Hadoop的安装都较为复杂
         正式安装软件之前给虚拟机做一个快照
         按照讲义中指定的软件安装
         按照讲义的步骤执行对应的命令，命令的遗漏会对后面的安装造成影响
   
   Airflow 视频录制的时候选择的是最新版本，现在Airflow版本升级了，讲义中的安装
步骤适合Airflow 1.10.11。
   请注意以下两条命令：
   -- 下载的时候指定 airflow 的版本
   pip install apache-airflow==1.10.11 -i https://pypi.douban.com/simple
   -- 下载的时候指定 mysqlclient 的版本
   pip install mysqlclient==1.4.6
   2).Python环境准备
   备注：提前下载 Python-3.6.6.tgz
   备注：使用linux122安装
   
   # 卸载 mariadb
   rpm -qa | grep mariadb
   mariadb-libs-5.5.65-1.el7.x86_64
   mariadb-5.5.65-1.el7.x86_64
   mariadb-devel-5.5.65-1.el7.x86_64
   yum remove mariadb
   yum remove mariadb-libs
   
   # 安装依赖
   rpm -ivh mysql57-community-release-el7-11.noarch.rpm
   yum install readline readline-devel -y
   yum install gcc -y
   yum install zlib* -y
   yum install openssl openssl-devel -y
   yum install sqlite-devel -y
   yum install python-devel mysql-devel -y
   
   # 提前到python官网下载好包
   cd /opt/lagou/software
   tar -zxvf Python-3.6.6.tgz
   
   # 安装 python3 运行环境
   cd Python-3.6.6/
   # configure文件是一个可执行的脚本文件。如果配置了--prefix，安装后的所有
   # 资源文件都会放在目录中
   ./configure --prefix=/usr/local/python3.6
   make && make install
   /usr/local/python3.6/bin/pip3 install virtualenv
   
   # 启动 python3 环境
   cd /usr/local/python3.6/bin/
   ./virtualenv env  
   . env/bin/activate
   
   # 检查 python 版本
   python -V
   3).安装Airflow
   # 设置目录（配置文件）
   # 添加到配置文件/etc/profile。未设置是缺省值为 ~/airflow
   export AIRFLOW_HOME=/opt/lagou/servers/airflow
   # 使用豆瓣源非常快。-i: 指定库的安装源（可选选项）
   pip install apache-airflow -i https://pypi.douban.com/simple
   pip install apache-airflow==1.10.11 -i https://pypi.douban.com/simple
   
   备注：
        apache-airflow==1.10.11,需要指定安装的版本，重要！！！
        软件安装路径在$AIRFLOW_HOME（缺省为~/airflow），此时目录不存在
        安装的是版本是1.10.11,不指定下载源时下载过程非常慢
   4).创建数据库用户并授权
   -- 创建数据库
   create database airflowlinux122;
   -- 创建用户airflow，设置所有ip均可以访问
   create user 'airflow'@'%' identified by '12345678';
   create user 'airflow'@'localhost' identified by '12345678';
   -- 用户授权，为新建的airflow用户授予Airflow库的所有权限
   grant all on airflowlinux122.* to 'airflow'@'%';
   SET GLOBAL explicit_defaults_for_timestamp = 1;
   flush privileges;
   5).修改Airflow DB配置
   # python3 环境中执行
   pip install mysqlclient==1.4.6
   airflow initdb
   备注：
        mysqlclient==1.4.6,需要指定安装的版本，重要！！！
   有可能在安装完Airflow找不到 $AIRFLOW_HOME/airflow.cfg 文件，执行完
airflow initdb才会在对应的位置找到该文件。
   勘误开始（2021年5月）
   在执行 airflow initdb 命令时，如遇上如下报错：
   ModuleNotFoundError: No module named 'sqlalchemy.ext.declarative.clsregistry'
   这是由于 SQLAlchemy 模块版本低导致的错误。执行以下命令后，重新执行
airflow initdb 命令。
   pip install SQLAlchemy==1.3.23 
   勘误结束（2021年5月）
   修改 $AIRFLOW_HOME/airflow.cfg：
   # 约 75 行
   sql_alchemy_conn = mysql://airflow:12345678@linux123:3306/airflowlinux122
   # 重新执行
   airflow initdb
   可能出现的错误：Exception: Global variable explicit_defaults_for_timestamp
needs to be on (1) for mysql
   解决方法：
   SET GLOBAL explicit_defaults_for_timestamp = 1;
   FLUSH PRIVILEGES;
   6).安装密码模块
   安装password组件:
   pip install apache-airflow[password] 
   修改 airflow.cfg 配置文件（第一行修改，第二行增加）：
   # 约 281 行
   约464行
   [webserver]
   # 约 353行
   约540行
   authenticate = True
   auth_backend = airflow.contrib.auth.backends.password_auth
   添加密码文件
   python命令，执行一遍；添加用户登录，设置口令

import airflow
from airflow import models, settings
from airflow.contrib.auth.backends.password_auth import PasswordUser

user = PasswordUser(models.User())
user.username = 'airflow'
user.email = 'airflow@lagou.com'
user.password = 'airflow123'

session = settings.Session()
session.add(user)
session.commit()
session.close()
exit()
   
   airflow users create --role Admin --firstname airflow --lastname airflow 
--password airflow123 --username tc --email  airflow@lagou.com
   7).启动服务
   # 备注：要先进入python3的运行环境
   cd /usr/local/python3.6/bin/
   ./virtualenv env  
   . env/bin/activate
   # 退出虚拟环境命令
   deactivate
   # 启动scheduler调度器:
   airflow scheduler -D
   # 服务页面启动：
   airflow webserver -D
   备注：airflow命令所在位置：/usr/local/python3.6/bin/env/bin/airflow
   安装完成，可以使用浏览器登录 linux122:8080,输入用户名、口令：airflow /airflow123
   8).修改时区
   Airflow默认使用UTC时间，在中国时区需要用+8 小时。将UTC修改为中国时区，需
要修改Airflow源码。
   (1).在修改 $AIRFLOW_HOME/airflow.cfg 文件
   # 约 65 行
   约 19 行
   default_timezone = Asia/Shanghai
   (2).修改 timezone.py
   # 进入Airflow包的安装位置
   cd /usr/local/python3.6/bin/env/lib/python3.6/site-packages/
   
   # 修改airflow/utils/timezone.py
   cd airflow/utils
   vi timezone.py
   
   第27行注释，增加29-37行：
utc = pendulum.timezone('UTC')

from airflow import configuration as conf
try:
        tz = conf.get("core", "default_timezone")
        if tz == "system":
                 utc = pendulum.local_timezone()
        else:
                 utc = pendulum.timezone(tz)
except Exception:
        pass
	
	备注：以上的修改方式有警告，可以使用下面的方式（推荐）：
utc = pendulum.timezone('UTC')

from airflow import configuration 
try:
        tz = configuration.conf("core", "default_timezone")
        if tz == "system":
                 utc = pendulum.local_timezone()
        else:
                 utc = pendulum.timezone(tz)
except Exception:
        pass
	
	修改utcnow()函数 (注释掉72行，增加73行内容)

62 def utcnow():
  63   """
  64   Get the current date and time in UTC
  65
  66   :return:
  67   """
  68
  69   # pendulum utcnow() is not used as that sets a
TimezoneInfo object
  70   # instead of a Timezone. This is not pickable and
also creates issues
  71   # when using replace()
  72   # d = dt.datetime.utcnow()
  73   d = dt.datetime.now()
  74   d = d.replace(tzinfo=utc)
  75
  76   return d
   (3).修改 airflow/utils/sqlalchemy.py
   # 进入Airflow包的安装位置
   cd /usr/local/python3.6/bin/env/lib/python3.6/site-packages/

   # 修改 airflow/utils/sqlalchemy.py
   cd airflow/utils
   vi sqlalchemy.py
   
   在38行之后增加 39 - 47 行的内容：

utc = pendulum.timezone('UTC')
from airflow import configuration as conf
try:
        tz = conf.get("core", "default_timezone")
        if tz == "system":
                utc = pendulum.local_timezone()
        else:
                utc = pendulum.timezone(tz)
except Exception:
        pass
	
	备注：以上的修改方式有警告，可以使用下面的方式（推荐）：
utc = pendulum.timezone('UTC')

from airflow import configuration 
try:
        tz = configuration.conf("core", "default_timezone")
        if tz == "system":
                 utc = pendulum.local_timezone()
        else:
                 utc = pendulum.timezone(tz)
except Exception:
        pass	
	
   (4).修改airflow/www/templates/admin/master.html
   # 进入Airflow包的安装位置
   cd /usr/local/python3.6/bin/env/lib/python3.6/site-packages/
   
   # 修改 airflow/www/templates/admin/master.html
   cd airflow/www/templates/admin
   vi master.html   
   
   # 将第40行修改为以下内容：
  40   var UTCseconds = x.getTime();
   # 将第43行修改为以下内容：
  43     "timeFormat":"H:i:s",
   
   重启airflow webserver
   
   # 关闭 airflow webserver 对应的服务
ps -ef | grep 'airflow-webserver' | grep -v 'grep' | awk
'{print $2}' | xargs -i kill -9 {}

   # 关闭 airflow scheduler 对应的服务
ps -ef | grep 'airflow' | grep 'scheduler' | awk '{print $2}'
| xargs -i kill -9 {}
   
   # 删除对应的pid文件
   cd $AIRFLOW_HOME
   rm -rf *.pid
   
   # 重启服务（在python3.6虚拟环境中执行）
   airflow scheduler -D
   airflow webserver -D
   9).Airflow的web界面
   Trigger Dag：人为执行触发
   Tree View：当dag执行的时候，可以点入，查看每个task的执行状态（基于树状视
图）。状态：success、running、failed、skipped、retry、queued、no status
   Graph View：基于图视图（有向无环图），查看每个task的执行状态
   Tasks Duration：每个task的执行时间统计，可以选择最近多少次执行
   Task Tries：每个task的重试次数
   Gantt View：基于甘特图的视图，每个task的执行状态
   Code View：查看任务执行代码
   Logs：查看执行日志，比如失败原因
   Refresh：刷新dag任务
   Delete Dag：删除该dag任务
   10).禁用自带的DAG任务
   停止服务：
   # 关闭 airflow webserver 对应的服务
ps -ef | grep 'airflow-webserver' | grep -v 'grep' | awk
'{print $2}' | xargs -i kill -9 {}
   
   # 关闭 airflow scheduler 对应的服务
ps -ef | grep 'airflow' | grep 'scheduler' | awk '{print $2}' |
xargs -i kill -9 {}
   # 删除对应的pid文件
   cd $AIRFLOW_HOME
   rm -rf *.pid
   
   修改文件 $AIRFLOW_HOME/airflow.cfg：
   # 修改文件第 136 行
  136 # load_examples = True
  137 load_examples = False
 
   # 重新设置db
   airflow resetdb -y
   
   重新设置账户、口令：

import airflow 
from airflow import models, settings
from airflow.contrib.auth.backends.password_auth import PasswordUser

user = PasswordUser(models.User())
user.username = 'airflow'
user.email = 'airflow@lagou.com'
user.password = 'airflow123'

session = settings.Session()
session.add(user)
session.commit()
session.close()
exit()
   
   重启服务
   # 重启服务
   airflow scheduler -D
   airflow webserver -D
   11).crontab简介
   Linux 系统则是由 cron (crond) 这个系统服务来控制的。Linux 系统上面原本就有非
常多的计划性工作，因此这个系统服务是默认启动的。
   
   Linux 系统也提供了Linux用户控制计划任务的命令：crontab 命令。
   日志文件：ll /var/log/cron*
   编辑文件：vim /etc/crontab
   进程：ps -ef | grep crond ==> /etc/init.d/crond restart
   作用：任务（命令）定时调度（如：定时备份，实时备份）
   简要说明：cat /etc/crontab

# For details see man 4 crontabs

# Example of job definition:
# .---------------- minute (0 - 59)
# |  .------------- hour (0 - 23)
# |  |  .---------- day of month (1 - 31)
# |  |  |  .------- month (1 - 12) OR jan,feb,mar,apr ...
# |  |  |  |  .---- day of week (0 - 6) (Sunday=0 or 7) OR sun,mon,tue,wed,thu,fri,sat
# |  |  |  |  |
# *  *  *  *  * user-name  command to be executed
    
   在以上各个字段中，还可以使用以下特殊字符：
   * 代表所有的取值范围内的数字。如月份字段为*，则表示1到12个月；
   / 代表每一定时间间隔的意思。如分钟字段为*/10，表示每10分钟执行1次；
   - 代表从某个区间范围，是闭区间。如2-5 表示2,3,4,5,小时字段中0-23/2 表示在
0~23 点范围内每2个小时执行一次；
   , 分散的数字（不连续）。如1,2,3,4,7,9；
   注：由于各个地方每周第一天不一样，因此Sunday=0(第1天）或Sunday=7(最后1天)。
   
   crontab配置实例
   # 每一分钟执行一次command（因cron默认每1分钟扫描一次，因此全为*即可）
   *  *  *  *  * command
   # 每小时的第3和第15分钟执行command
   3,15  *  *  *  * command
   # 每天上午8-11点的第3和15分钟执行command
   3,15  8-11 * * * command
   # 每隔2天的上午8-11点的第3和15分钟执行command
   3,15  8-11 */2 *  * command
   # 每个星期一的上午8点到11点的第3和第15分钟执行command
   3,15  8-11  *  *  1 command
   # 每晚的21:30执行command
   30  21  *  * * command
   # 每月1、10、22日的4:45执行command
   45  4  1,10,22 * * command
   # 每周六、周日的1 : 10执行command
   10  1 * *  6,0 command
   # 每小时执行command
   0 */1  *  *  * command
   # 晚上11点到早上7点之间，每隔一小时执行command
   *  23-7/1 *  *  * command
   
   